package org.yagnus.yadoop;
import org.apache.hadoop.fs.FileStatus
import org.apache.hadoop.fs.PathFilter
import org.apache.hadoop.fs.FileSystem
import scala.collection.mutable.ArrayBuffer
import org.apache.hadoop.fs.Path
import org.yagnus.yadoop.Yadoop._
import org.apache.hadoop.fs.permission.FsAction
import java.util.regex.Pattern

object HdfsUtils {
  def getFileDirStatus(fs: FileSystem, inFile: String, regExp: String, recursive: Boolean): Tuple2[List[FileStatus], List[FileStatus]] =
    getFileDirStatus(fs, Seq(inFile), regExp, recursive);

  def getFileDirStatus(fs: FileSystem, inFiles: Seq[String], regExp: String, recursive: Boolean): Tuple2[List[FileStatus], List[FileStatus]] = {
    val pat =
      if (regExp == null)
        Pattern.compile(".*");
      else
        Pattern.compile(regExp);

    def accept(p: Path) = {
      //      println("HdfsUtils checking " + pathStrOfPath(p) + " for match " + pat.matcher(pathStrOfPath(p)).matches);
      pat.matcher(pathStrOfPath(p)).matches;
    }
    var toBeChecked = List() ++ inFiles;
    var inputDirectories = List[FileStatus]();
    var inputFileStata = List[FileStatus]();

    while (!toBeChecked.isEmpty) {
      val glob = toBeChecked.head;
      toBeChecked = toBeChecked.tail;
      //      println("Hdfs Checking " + glob);

      val matches = accept(glob);

      if (fs.exists(glob)) {
        //        println(" it exists in fs:" + glob);
        val fileS = fs.getFileStatus(glob);

        //recurse into director
        if (fileS.isDir) {
          if (recursive)
            toBeChecked :+= (glob + "/*");
          if (matches) inputDirectories :+= fileS;
        } else { //it's a file
          if (matches) inputFileStata :+= fileS;
        }
      } else { //not a file but a pattern
        //        println(" it doesn't exist in fs:" + glob);
        val results = fs.globStatus(glob);
        if (results != null && !results.isEmpty) {
          for (flstts ← results) {
            val readPerm = flstts.getPermission.getUserAction.implies(FsAction.READ)
            if (!readPerm) {
              //              println("Skipping a file because it is not readable.");
            } else {
              if (flstts.isDir) {
                if (accept(flstts))
                  inputDirectories :+= flstts;
                if (recursive)
                  toBeChecked :+= pathStrOfFileStatus(flstts) + "/*";
              } else {
                if (accept(flstts))
                  inputFileStata :+= flstts;
              }

            }
          }
        } /* else {
        println("HdfsUtils: empty result.");
      }*/
      }
    }
    //TODO: sort the directories and files so that it is advantageous to PieceMaker;

    return (inputFileStata, inputDirectories);

  }
  /**
   * Reads the content of a file fully into byte array. returns null on error.
   */
  def readFully(fs: FileSystem, fn: String): Array[Byte] = readFully(fs, new Path(fn));
  def readFully(fs: FileSystem, fp: Path): Array[Byte] = {
    try {
      val fileStat = fs.getFileStatus(fp);
      val fileLen: Int = fileStat.getLen.asInstanceOf[Int];
      //      println("The file at " + fileStat.getPath.toUri.getPath + " has size " + fileLen + ", the block size is :" + fileStat.getBlockSize());
      val file = fs.open(fp);
      val ret = new Array[Byte](fileLen);
      file.readFully(ret);
      return ret;
    } catch {
      case _ => return null;
    }
  }

  def cleanMROutputDir(fs: FileSystem, odir: String) = {
    val dl = fs.delete(odir + "/_logs", true);
    val ds = fs.delete(odir + "/_SUCCESS", true);
    dl && ds;
  }
}